// Package order
// Author liuzhiyong
// Date 2023/11/21
// Description 业务层代码
package order

import (
	"context"
	"encoding/json"
	"fmt"
	"gitee.com/liu_zy_developer/go_order_service/config"
	"gitee.com/liu_zy_developer/go_order_service/dao/mq"
	"gitee.com/liu_zy_developer/go_order_service/dao/mysql"
	"gitee.com/liu_zy_developer/go_order_service/model"
	"gitee.com/liu_zy_developer/go_order_service/proto"
	"gitee.com/liu_zy_developer/go_order_service/rpc"
	"gitee.com/liu_zy_developer/go_order_service/third_party/snowflake"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
	"strconv"
)

// OrderEntity 自定义结构体，实现了两个方法, 本地事务的方法和查询本地事务的方法
// 发送事务消息的时候 RocketMQ 会自动根据情况调用那两个方法
type OrderEntity struct {
	OrderId int64           // 订单号
	Param   *proto.OrderReq // RPC远程调用的参数
	err     error           // 异常信息, 记录本地事务是否执行成功
}

// ExecuteLocalTransaction 执行本地事务
// 当发送half-message成功之后, 这个方法就会被执行
func (o *OrderEntity) ExecuteLocalTransaction(*primitive.Message) primitive.LocalTransactionState {
	fmt.Println("in ExecuteLocalTransaction...")
	// 校验参数, 用于执行RPC调用的参数为空, 必要参数为空, 不能执行后续流程
	if o.Param == nil {
		zap.L().Error("ExecuteLocalTransaction param is nil")
		o.err = status.Error(codes.Internal, "invalid OrderEntity") // 给结构体中的异常赋值, 调用方可以拿到, 用于判断本地是否执行是否成功, 有没有出现问题
		return primitive.CommitMessageState                         // 这里感觉取消MQ的消息也可以, 后续代码都还没有执行
	}
	param := o.Param
	ctx := context.Background()
	// 1. 查询商品金额(正常情况是请求营销中心, 但是本示例没有营销中心) ==> RPC连接 goods_service
	goodsDatail, err := rpc.GoodsCli.GetGoodsDetail(ctx, &proto.GetGoodsDetailReq{
		GoodsId: param.GoodsId,
		UserId:  param.UserId,
	})
	if err != nil {
		zap.L().Error("GoodsCli.GetGoodsDetail failed", zap.Error(err))
		// 查询商品价格失败
		o.err = status.Error(codes.Internal, err.Error())
		// 此时此次事务中的所有远程调用还未执行操作, 哪怕是本次操作, 因为远程返回错误, 远程的本地事务也会回滚, 查询请求也不涉及事务, MQ没必要发送回滚消息
		return primitive.RollbackMessageState
	}

	payAmountStr := goodsDatail.Price
	payAmount, _ := strconv.ParseInt(payAmountStr, 10, 64)

	// 2. 库存校验以及扣减 ==> RPC连接 stock_service
	_, err = rpc.StockCli.ReduceStock(ctx, &proto.GoodsStockInfo{
		OrderId: o.OrderId,
		GoodsId: o.Param.GoodsId,
		Num:     o.Param.Num,
	})
	if err != nil {
		// 库存扣减失败，丢弃half-message, 上一步操作是查询操作, 不涉及回滚事务
		zap.L().Error("StockCli.ReduceStock failed", zap.Error(err))
		o.err = status.Error(codes.Internal, "ReduceStock failed")
		return primitive.RollbackMessageState // 丢弃半消息
	}

	// 代码能执行到这里说明 扣减库存成功了，
	// 从这里开始如果本地事务执行失败就需要回滚库存
	// 3. 创建订单
	// 创建订单对象
	orderData := model.Order{
		OrderId:        o.OrderId,
		UserId:         param.UserId,
		PayAmount:      payAmount,
		ReceiveAddress: param.Address,
		ReceiveName:    param.Name,
		ReceivePhone:   param.Phone,
		Status:         100, // 待支付
	}
	// 创建订单详情对象
	orderDetail := model.OrderDetail{
		OrderId: o.OrderId,
		UserId:  param.UserId,
		GoodsId: param.GoodsId,
		Num:     param.Num,
	}
	// 在本地事务创建订单和订单详情记录
	err = mysql.CreateOrderWithTransaction(ctx, &orderData, &orderDetail)
	if err != nil {
		// 本地事务执行失败了 ==> 需要回滚上面的操作
		// 本地事务执行失败了，上一步已经库存扣减成功
		// 就需要将库存回滚的消息投递出去，小游根据消息进行库存回滚
		zap.L().Error("CreateOrderWithTransation failed", zap.Error(err))
		return primitive.CommitMessageState // 将之前发送的hal-message 提交, 下游系统收到消息之后会进行回滚, 这一步不需要记录error了, 也就是给o.err赋值
	}

	// 发送延迟消息 --> 放在这里不合适, 提到外面去
	// 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
	data := model.OrderGoodsStockInfo{
		OrderId: o.OrderId,
		GoodsId: param.GoodsId,
		Num:     param.Num,
	}
	b, _ := json.Marshal(data)
	msg := primitive.NewMessage("xx_pay_timeout", b)
	msg.WithDelayTimeLevel(3) // 设置延时等级
	_, err = mq.Producer.SendSync(context.Background(), msg)
	if err != nil {
		// 发送延时消息失败
		zap.L().Error("send delay msg failed", zap.Error(err))
		return primitive.CommitMessageState
	}

	// 走到这里说明本地事务执行成功 ==> 丢弃掉MQ中的半消息
	return primitive.RollbackMessageState
}

// CheckLocalTransaction 检查本地事务
// 当hale-message 没有响应时, broker 会回查本地事务的状态, 此时这个方法会被执行
func (o *OrderEntity) CheckLocalTransaction(*primitive.MessageExt) primitive.LocalTransactionState {
	// 检查本地状态是否创建成功订单
	_, err := mysql.QueryOrder(context.Background(), o.OrderId)
	// 需要再查询订单详情表
	if err == gorm.ErrRecordNotFound {
		// 没查询到说明订单创建失败，需要回滚库存
		return primitive.CommitMessageState
	}
	return primitive.RollbackMessageState
}

// Create 创建订单 利用RocketMQ的半消息实现分布式事务
func Create(ctx context.Context, param *proto.OrderReq) error {
	// 1. 生成单号
	orderId := snowflake.GenID()

	//2. 发送事务half-message
	// 2.1 创还能事务监听器, 用于执行本地事务
	orderEntity := &OrderEntity{
		OrderId: orderId,
		Param:   param,
	}
	// 2.2 创建事务消息生产者
	p, err := rocketmq.NewTransactionProducer(
		orderEntity, // 事务监听器, 用于执行本地事务
		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})), // 配置nameserver
		producer.WithRetry(2),                 // 重试次数
		producer.WithGroupName("order_srv_1"), // 生产者组
	)
	if err != nil {
		zap.L().Error("NewTransactionProducer failed", zap.Error(err))
		return status.Error(codes.Internal, "NewTransactionProducer failed") // gPRC的错误响应
	}
	// 2.3 启动生产者
	p.Start()

	// 2.4 封装回滚消息 orderId GoodsId num
	data := model.OrderGoodsStockInfo{
		OrderId: orderId,
		GoodsId: param.GoodsId,
		Num:     param.Num,
	}
	b, _ := json.Marshal(data)
	msg := &primitive.Message{
		Topic: config.Conf.RocketMqConfig.Topic.StockRollback, // xx_stock_rollback
		Body:  b,
	}
	// 2.5 发送事务消息
	res, err := p.SendMessageInTransaction(context.Background(), msg)
	if err != nil {
		zap.L().Error("SendMessageInTransaction failed", zap.Error(err))
		return status.Error(codes.Internal, "create order failed")
	}
	zap.L().Info("p.SendMessageInTransaction success", zap.Any("res", res))

	// 2.6 校验事务消息是否发送成功
	// 如果回滚库存的消息被投递出去（commit）说明本地事务里面手动出发了回滚操作，也就是创建订单失败
	if res.State == primitive.CommitMessageState {
		return status.Error(codes.Internal, "create order failed")
	}

	// 3 校验本地事务是否执行成功
	// 其他内部错误
	if orderEntity.err != nil {
		return orderEntity.err
	}
	return nil
}
